package com.hr.reactive.operators;

import com.hr.reactive.common.Utils;
import com.hr.reactive.mono.s02.Lec04MonoEmptyTest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;

/**
 * @author huran
 *  do hook callbacks
 */
public class Lec03DoCallbacks {
    private static final Logger log = LoggerFactory.getLogger(Lec03DoCallbacks.class);
    public static void main(String[] args) {
    //下游订阅了2个，create方法生产者也不生产4个不影响
       Flux.create(fluxSink -> {
            log.warn("producer begin");
           for (int i = 0; i < 4; i++) {
               fluxSink.next(i);
           }
           // do方法不会改变值
           fluxSink.complete();
          // fluxSink.error(new RuntimeException("test"));
           log.warn("producer done");
       }).
               doOnComplete(()->{log.warn("doOnComplete-1");}).
               //开始之前调用
               doFirst(()->{log.warn("doFirst-1");}).
               //每个元素发出时 会调用这个方法
               doOnNext(item->log.warn("doOnNext-1:{}",item))
               //向下游发送订阅对象时，调用附带订阅对象，
               .doOnSubscribe(subscription -> log.warn("doOnSubscribe-1:{}",subscription))
               //当下游向发送者发送请求时，在请求时调用该方法
               .doOnRequest(req-> log.warn("doOnRequest-1:{}",req))
               //当生产者发生错误时调用该方法
               .doOnError(err->log.warn("doOnError-1: ",err))
               //不管是完成是，还是发生错误 都会调用该方法
               .doOnTerminate(()->log.warn("doOnTerminate-1"))
               //取消时调用
               .doOnCancel(()->log.warn("doOnCancel-1"))
               //生产者发送数据 订阅者没有收到 调用该方法
               .doOnDiscard(Object.class,o->log.warn("doOnDiscard-1:{}",o))
               //无论发生什么都会执行
               .doFinally(signalType -> log.warn("doFinally-1:{}",signalType))

               .take(20)
               .doOnComplete(()->{log.warn("doOnComplete-2");})
               .doFirst(()->{log.warn("doFirst-2");})
               .doOnNext(item->log.warn("doOnNext-2:{}",item))
               .doOnSubscribe(subscription -> log.warn("doOnSubscribe-2:{}",subscription))
               .doOnRequest(req-> log.warn("doOnRequest-2:{}",req))
               .doOnError(err->log.warn("doOnError-2:",err))
               .doOnTerminate(()->log.warn("doOnTerminate-2"))
               .doOnCancel(()->log.warn("doOnCancel-2"))
               .doOnDiscard(Object.class,o->log.warn("doOnDiscard-2:{}",o))
               .doFinally(signalType -> log.warn("doFinally-2:{}",signalType))
               .take(4)
               .subscribe(Utils.subscriber("subscriber"));
   }
}
